package com.gmall.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.gmall.utils.MykafkaUtil;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.text.SimpleDateFormat;

public class UniqueVisitorApp {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //todo 读取kafka主数据
        String groupId = "unique_visit_app";
        String sourceTopic = "dwd_page_log";//记录用户登录的log日志
        String sinkTopic = "dwm_unique_visit";
        DataStreamSource<String> kafkaDS = env.addSource(MykafkaUtil.getKafkaConsumer(sourceTopic, groupId));

        //将每行数据转换成json对象
        SingleOutputStreamOperator<JSONObject> jsonObj = kafkaDS.map(JSON::parseObject);

        //todo 过滤数据 状态编程 只保留每个mid每天第一次的登入时间
        KeyedStream<JSONObject, String> keyedStream = jsonObj.keyBy(j -> j.getJSONObject("common").getString("mid"));
        SingleOutputStreamOperator<JSONObject> uvDS = keyedStream.filter(new RichFilterFunction<JSONObject>() {
            private ValueState<String> dataState;
            private SimpleDateFormat simpleDateFormat;

            public void open(Configuration parameters) {
                ValueStateDescriptor<String> stringValueStateDescriptor = new ValueStateDescriptor<>("data-state", String.class);
                //设置状态的超时时间以及更新时间的方式
                StateTtlConfig stateTtlConfig = new StateTtlConfig.Builder(Time.hours(24)).setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite).build();
                stringValueStateDescriptor.enableTimeToLive(stateTtlConfig);
                dataState = getRuntimeContext().getState(stringValueStateDescriptor);

                simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd");
            }

            @Override
            public boolean filter(JSONObject jsonObject) throws Exception {
                //取出上一条页面信息
                String lastPageid = jsonObject.getJSONObject("page").getString("last_page_id");
                //判断上一条页面是否为null
                if (lastPageid == null || lastPageid.length() <= 0) {
                    //取出状态数据
                    String lastDate = dataState.value();

                    //取出今天的日期
                    String curDate = simpleDateFormat.format(jsonObject.getLong("ts"));

                    //判断两个日期是否相同
                    if (!curDate.equals(lastDate)) {
                        dataState.update(curDate);
                        return true;
                    }
                }
                return false;
            }
        });
        //将数据写入kafka
        uvDS.print("uv>>>>>>>>");
        uvDS.map(JSONAware::toJSONString).addSink(MykafkaUtil.getKafkaProducer(sinkTopic));

        //启动任务
        env.execute("UniqueVisitApp");
    }
}
